package com.wei.weblog.pre;

import java.io.IOException;
import java.util.HashSet;
import java.util.Set;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import com.wei.weblog.mr.WebLogInfo;
import com.wei.wordcount.WordCountJob;
import com.wei.wordcount.WordCountMap;
import com.wei.wordcount.WordCountReduce;

/**
 * 网站日志清洗
 * @author Administrator
 *
 */
public class WebLogDataCleanJob {

	static class WebLogDataCleanMapper extends Mapper<LongWritable, Text, Text, NullWritable>{

		Text text=new Text();
		NullWritable v= NullWritable.get();
		Set<String> pages=new HashSet<>();
		@Override
		protected void setup(Context context) throws IOException,
				InterruptedException {
			pages.add("/about");
			pages.add("/black-ip-list/");
			pages.add("/cassandra-clustor/");
			pages.add("/finance-rhive-repurchase/");
			pages.add("/hadoop-family-roadmap/");
			pages.add("/hadoop-hive-intro/");
			pages.add("/hadoop-zookeeper-intro/");
			pages.add("/hadoop-mahout-roadmap/");
		}
		
		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			String line = value.toString();
			WebLogInfo webLogInfo=WebLogParser.parser(line);
			WebLogParser.filterUrl(webLogInfo,pages);
			if (webLogInfo.isValid()) {
				text.set(webLogInfo.toString());
				context.write(text,v);
			}
		}

	}
	
	public static void main(String[] args) throws Exception {
		Configuration configuration = new Configuration();
		Job job = Job.getInstance(configuration);
		job.setJarByClass(WebLogDataCleanJob.class);

		job.setMapperClass(WebLogDataCleanMapper.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(NullWritable.class);

		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(NullWritable.class);

		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		boolean waitForCompletion = job.waitForCompletion(true);
		System.exit(waitForCompletion ? 0 : 1);
	}
	
	
}
